package org.apache.zookeeper.server.paxos2;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import org.apache.log4j.Logger;

public class SocketManager extends Thread {

	private static final Logger LOG = Logger.getLogger(Storage.class);
	private final int port = 5454;
	private ServerSocket ss = null;

	private Storage storage;

	private Socket[] outgoing = new Socket[Configuration.servers.length];
	private DataOutputStream[] outgoingStream = new DataOutputStream[Configuration.servers.length];

	public SocketManager() {
		try {
			ss = new ServerSocket(port);
		} catch (IOException e) {
			LOG.warn("create server socket", e);
		}
	}

	public void setStorage(Storage storage){
	    	this.storage = storage;
	}

	private static int getSid(Socket sock) {
		String host = sock.getInetAddress().getHostName();
		for (int i = 0; i < Configuration.servers.length; i++) {
			if (host.equals(Configuration.servers[i]))
				return i;
		}
		System.out.println("unknown host " + host);
		return -1;
	}

	public void run() {
		while (true) {
			try {
				Socket sock = ss.accept();
				int sid = getSid(sock);
				if (sid != -1) {
					new Receiver(sock, sid).start();
				}
			} catch (IOException e) {
				LOG.warn("accept error", e);
				break;
			}
		}
	}

	private class Receiver extends Thread {
		private Socket sock = null;
		private int sid;

		public Receiver(Socket sock, int sid) {
			System.out.println("Receive a socket from "+sock);
			this.sock = sock;
			this.sid = sid;
		}

		public void run() {
			try {
				DataInputStream dis = new DataInputStream(sock
						.getInputStream());
				while (true) {
					System.out.println("waiting on message");
					Message msg = Message.read(dis);
					System.out.println("receive a msg "+msg);
					storage.handleMessage(msg, sid);
				}
			} catch (Exception e) {
				LOG.warn("receive error", e);
				synchronized(this){
				    outgoing[sid]=null;
				    outgoingStream[sid]=null;
				}
			}
		}
	}

	public synchronized boolean send(int id, Message msg) {
		try {
			if (outgoing[id] == null) {
				outgoing[id] = new Socket(Configuration.servers[id], port);
				outgoing[id].setTcpNoDelay(true);
				outgoingStream[id] = new DataOutputStream(new BufferedOutputStream(outgoing[id].getOutputStream()));
			}
			Message.write(msg, outgoingStream[id]);
			outgoingStream[id].flush();
			return true;
		} catch (Exception e) {
			LOG.warn("send error", e);
			outgoing[id] = null;
			outgoingStream[id] = null;
			return false;
		}
	}

}
